package com.giraffe.device.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.giraffe.device.api.DeviceService;
import com.giraffe.device.config.InfluxDBConfig;
import com.giraffe.device.pojo.dto.DeviceDTO;
import com.giraffe.device.strategy.DeviceMessageParsingStrategy;
import com.giraffe.device.strategy.SecondaryDeviceMessageParsingStrategy;
import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.WriteApiBlocking;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.Instant;
import java.util.List;

/**
 * @Author ldx
 * @Date 2024/01/21 23:42
 **/
@Slf4j
@Component
public class MqttTopicListener {

    @Resource
    private InfluxDBClient influxDBClient;
    @Resource
    private InfluxDBConfig influxDBConfig;
    @Resource
    private List<DeviceMessageParsingStrategy> deviceMessageParsingStrategies;
    @Resource
    private DeviceService deviceService;

    /**
     * 消息消费者
     *
     * @param consumerRecord 消息记录
     */
    @KafkaListener(topics = {"mqtt_data_topic_up"},groupId="giraffe_mqtt_group_up")
    public void consumer(ConsumerRecord<?, ?> consumerRecord) {
        log.info("----消息处理---");
        log.info("收到bigData推送的数据'{}'", consumerRecord.toString());
        JSONObject obj = JSON.parseObject(consumerRecord.value().toString());

        WriteApiBlocking writeApi = influxDBClient.getWriteApiBlocking();
        Point point = Point
                .measurement("device_up_message")
                .addTag("deviceId", obj.getString("deviceId"))
                .addTag("topic", obj.getString("topic"))
                .addField("message", obj.getString("message"))
                .time(Instant.now(), WritePrecision.NS);
        writeApi.writePoint(influxDBConfig.getBucket(), influxDBConfig.getOrg(), point);

        DeviceDTO dto = deviceService.getDevice(obj.getString("deviceId"));
        log.info("----设备信息---{}",JSON.toJSONString(dto));
        DeviceMessageParsingStrategy parsingStrategy = deviceMessageParsingStrategies.stream().filter(f -> f.deviceTypeCode().equals(dto.getDeviceType())).findFirst().get();
        parsingStrategy.parsing(obj.getString("deviceId"),obj.getString("topic"),obj.getString("message"));
    }


}
